/*
 * Copyright 2020 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.zq.learn.springbatch.mongo;

import com.zq.learn.springbatch.mongo.sourcedata.SourceData;
import io.micrometer.core.instrument.Statistic;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.data.MongoItemReader;
import org.springframework.batch.item.data.MongoItemWriter;
import org.springframework.batch.item.data.builder.MongoItemReaderBuilder;
import org.springframework.batch.item.data.builder.MongoItemWriterBuilder;
import org.springframework.batch.item.support.CompositeItemProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoOperations;
import org.springframework.data.mongodb.core.MongoTemplate;
import sun.awt.AWTIcon32_security_icon_yellow16_png;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;

@Configuration
@EnableBatchProcessing
public class JobConfig {
  @Autowired
  private MongoTemplate mongoTemplate;
  @Autowired
	private JobBuilderFactory jobBuilderFactory;
  @Autowired
	private StepBuilderFactory stepBuilderFactory;


	@Bean
  @StepScope
	public MongoItemReader<SourceData> mongoItemReader(@Value("#{jobParameters[status]}") Long status) {
		Map<String, Sort.Direction> sortOptions = new HashMap<>();
		sortOptions.put("title", Sort.Direction.DESC);
		return new MongoItemReaderBuilder<SourceData>()
				.name("sourceItemReader")
				.collection("SourceData")
				.targetType(SourceData.class)
				.template(mongoTemplate)
				.jsonQuery("{status:{$eq:?0}}")
        .parameterValues(status)
        .pageSize(10)
				.sorts(sortOptions)
				.build();
	}

	@Bean
	public MongoItemWriter<StatisticsResult> mongoItemWriter() {
		return new MongoItemWriterBuilder<StatisticsResult>()
				.template(mongoTemplate)
				.collection("StatisticsResult")
				.build();
	}

  /**
   * 统计订单数量
   * @return
   */
	@Bean
	public ItemProcessor<SourceData, StatisticsResult> countProcessor(){
	  return new ItemProcessor<SourceData, StatisticsResult>() {
	    //数量累积
	    private int count;
      @Override public StatisticsResult process(SourceData sourceData) throws Exception {
        ++count;
        System.out.println("++++++++++++++++++++++count"+count);
        SimpleDateFormat sdf=new SimpleDateFormat("yyyyMMdd");
        String day = sdf.format(new Date());
        return new StatisticsResult("count_"+day,day,"count",count);
      }
    };
  }

  /**
   * 统计订单金额
   * @return
   */
  @Bean
  public ItemProcessor<SourceData, StatisticsResult> priceProcessor(){
    return new ItemProcessor<SourceData, StatisticsResult>() {
      //金额累积
      private int price;
      @Override public StatisticsResult process(SourceData sourceData) throws Exception {
        price+=sourceData.getPrice();
        System.out.println("++++++++++++++++++++++Price"+price);
        SimpleDateFormat sdf=new SimpleDateFormat("yyyyMMdd");
        String day = sdf.format(new Date());
        return new StatisticsResult("price_"+day,day,"price",price);
      }
    };
  }

	@Bean
	public Step stepCount(MongoItemReader<SourceData> mongoItemReader, MongoItemWriter<StatisticsResult> mongoItemWriter) {
		return this.stepBuilderFactory.get("stepCount")
				.<SourceData, StatisticsResult>chunk(10)
				.reader(mongoItemReader)
        .processor(countProcessor())
				.writer(mongoItemWriter)
				.build();
	}
  @Bean
  public Step stepPrice(MongoItemReader<SourceData> mongoItemReader, MongoItemWriter<StatisticsResult> mongoItemWriter) {
    return this.stepBuilderFactory.get("step")
      .<SourceData, StatisticsResult>chunk(10)
      .reader(mongoItemReader)
      .processor(priceProcessor())
      .writer(mongoItemWriter)
      .build();
  }

	@Bean
	public Job job(Step stepCount,Step stepPrice) {
		return this.jobBuilderFactory.get("job")
				.start(stepCount)
        .next(stepPrice)
				.build();
	}

}
